#include "bthread.h"

#include <thread>
#include <vector>
#include <future>
#include <functional>
// 全局原子计数器，用于统计处理的任务数
std::atomic<int> processed_count{0};
// 消费者线程函数
void consumer(BThread<int>& queue, int id) {
    while (true) {
        int task;

        // 等待最多 200ms 获取任务
        if (queue.wait_and_pop(task, 200ms)) {
            // 模拟任务处理
            std::this_thread::sleep_for(50ms);
            processed_count++;
            std::cout << "Consumer " << id << " processed task: " << task << std::endl;
        } else {
            // 超时，检查是否已关闭
            if (queue.is_closed() && queue.size() == 0) {
                break; // 安全退出
            }
            // 否则继续等待
        }
    }
    std::cout << "Consumer " << id << " exiting." << std::endl;
}
// 生产者：使用 async 异步提交任务
std::future<void> start_producer(BThread<int>& queue, int total_tasks) {
    return std::async(std::launch::async, [&queue, total_tasks]() {
        for (int i = 1; i <= total_tasks; ++i) {
            try {
                queue.push(i);
                std::this_thread::sleep_for(20ms); // 控制生产速度
            } catch (const std::exception& e) {
                std::cerr << "Producer error: " << e.what() << std::endl;
                break;
            }
        }
        std::cout << "Producer finished submitting " << total_tasks << " tasks." << std::endl;
    });
}
int mains() {
    try {
        BThread<int> task_queue;
        const int num_consumers = 3;
        const int total_tasks = 20;
        // 启动消费者线程
        std::vector<std::thread> consumers;
        for (int i = 0; i < num_consumers; ++i) {
            consumers.emplace_back(consumer, std::ref(task_queue), i + 1);
        }
        // 启动生产者（异步）
        auto producer_future = start_producer(task_queue, total_tasks);
        // 等待生产者完成
        producer_future.wait();
        // 等待所有任务被消费（可优化为计数器，此处简单等待）
        std::this_thread::sleep_for(1s);
        // 关闭队列，通知消费者退出
        task_queue.close();
        // 等待所有消费者线程结束
        std::for_each(consumers.begin(), consumers.end(), [](std::thread& t) {
            if (t.joinable()) {
                t.join();
            }
        });
        // 输出统计结果
        std::cout << "All tasks processed. Total processed: " << processed_count.load() << std::endl;
    } catch (const std::exception& e) {
        std::cerr << "Exception in main: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}
